package com.atguigu.edu.realtime.app.dwd.trade;

import com.atguigu.edu.realtime.app.BaseAppSql;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.common.KafkaTopicConfig;
import com.atguigu.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradePaySuc extends BaseAppSql {
    public static void main(String[] args) {
        new DwdTradePaySuc().init(3074,2,"DwdTradePaySuc");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(30));

        //读取订单表
        tEnv.executeSql("create table dwd_trade_order_detail(\n" +
                "    id string,\n" +
                "    order_id string,\n" +
                "    course_id string,\n" +
                "    course_name string,\n" +
                "    date_id string,\n" +
                "    create_time string,\n" +
                "    \n" +
                "    origin_amount string,\n" +
                "    coupon_reduce string,\n" +
                "    final_amount string,\n" +
                "    od_ts string,\n" +
                "    \n" +
                "    user_id string,\n" +
                "    province_id string,\n" +
                "    expire_time_date string,\n" +
                "    expire_time string,\n" +
                "    session_id string,\n" +
                "    oi_ts string,\n" +
                "    row_op_ts timestamp_ltz(3) \n" +
                ")" + SQLUtil.getConnectorKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,"DwdTradePaySuc"));
        //读取支付表
        sourceFromOdsDb(tEnv,"DwdTradePaySuc");

        Table paymentInfo = tEnv.sqlQuery("select \n" +
                "    `data`['id'] id,\n" +
                "    `data`['out_trade_no'] out_trade_no,\n" +
                "    `data`['order_id'] order_id,\n" +
                "    `data`['alipay_trade_no'] alipay_trade_no,\n" +
                "    `data`['total_amount'] total_amount,\n" +
                "    `data`['trade_body'] trade_body,\n" +
                "    `data`['payment_type'] payment_type,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['update_time'] update_time,\n" +
                "    `data`['callback_content'] callback_content,\n" +
                "    `data`['callback_time'] callback_time,\n" +
                "    `ts` \n" +
                "from ods_db \n" +
                "where `database` = 'edu' and `table` = 'payment_info' \n" +
                //"    and `old['payment_status']` is not null \n" +
                "    and `data`['payment_status'] = '1602'");
        tEnv.createTemporaryView("payment_info",paymentInfo);
        //两表join
        Table result = tEnv.sqlQuery("select \n" +
                "    pi.id,\n" +
                "    od.id order_detail_id,\n" +
                "    od.order_id,\n" +
                "    od.course_id,\n" +
                "    od.course_name,\n" +
                "    od.origin_amount,\n" +
                "    od.coupon_reduce,\n" +
                "    od.final_amount,\n" +
                "    od.user_id,\n" +
                "    od.province_id,\n" +
                "    pi.payment_type,\n" +
                "    pi.callback_time,\n" +
                "    pi.ts*1000\n" +
 //               "    row_op_ts \n" +
                "from payment_info pi join dwd_trade_order_detail od on pi.order_id = od.order_id");

        //创建动态表和Kafka dwd_trade_pay_suc 关联
        tEnv.executeSql("create table dwd_trade_pay_suc(\n" +
                "    id string,\n" +
                "    order_detail_id string,\n" +
                "    order_id string,\n" +
                "    course_id string,\n" +
                "    course_name string,\n" +
                "    origin_amount string,\n" +
                "    coupon_reduce string,\n" +
                "    final_amount string,\n" +
                "    user_id string,\n" +
                "    province_id string,\n" +
                "    payment_type string,\n" +
                "    callback_time string,\n" +
                "    ts bigint\n" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_PAY_SUC));
        //插入写出数据
        result.executeInsert("dwd_trade_pay_suc");

    }
}
